Problem

AbstractKafkaConsumer.init() forwards the entire stored options — including the connection-level kafka config — into @platformatic/kafka's Consumer.consume(). Inside consume(), the MessagesStream constructor runs structuredClone(otherOptions) on whatever it doesn't explicitly destructure. Anything function-valued reachable from the kafka config crashes that clone with DOMException [DataCloneError], taking down consumer init.

The most common trigger is AWS MSK IAM authentication, which uses SASL/OAUTHBEARER with an async token provider:

sasl: {
  mechanism: 'OAUTHBEARER',
  token: async () => (await generateAuthToken({ region })).token,
}

Production stack trace:

DOMException [DataCloneError]: async () => { ... } could not be cloned.
    at structuredClone (node:internal/worker/js_transferable:126:26)
    at new MessagesStream (.../@platformatic/kafka/dist/clients/consumer/messages-stream.js:142:25)
    at #performConsume (.../@platformatic/kafka/dist/clients/consumer/consumer.js:942:24)
    ...
    at AbstractKafkaConsumer.init (.../@message-queue-toolkit/kafka/dist/AbstractKafkaConsumer.js:112:19)

The same class of bug would hit any function-valued field passed through SASL/OAUTHBEARER providers (Confluent Cloud OIDC, etc.).

Fix

Exclude kafka from the spread that builds consumeOptions in init():

- const { handlers: _, ...consumeOptions } = this.options
+ const { handlers: _, kafka: __, ...consumeOptions } = this.options

kafka is connection-level config — it belongs on the Consumer constructor (where it's already spread separately), not on consume(). Stripping it makes the consume args structuredClone-safe regardless of what's inside the user's connection config.

Tests

Two regression tests in PermissionConsumer.spec.ts that spy on Consumer.prototype.consume:

• strips 'kafka' from consume options — asserts the kafka field is absent from the args.
• keeps 'kafka' out of consume options even when it carries a function-valued field — injects a function into the kafka config (simulating AWS MSK IAM sasl.token) and asserts it never reaches consume().
  Verified both fail on the unfixed code — the second test reproduces the exact production stack trace — and pass with the fix applied.

Summary by CodeRabbit

• Bug Fixes

  • Fixed Kafka consumer configuration handling to prevent serialization errors with authentication token providers.

• Tests

  • Added regression test for Kafka consumer configuration validation.